using KafkaFlow;
using KafkaFlow.Producers;
using KafkaFlow.Serializer;
using Microsoft.Extensions.DependencyInjection;

namespace Application.KafkaService;

public class KafkaProducerService
{
    // private readonly IProducer<Null, string> _producer;
    //
    // public KafkaProducerService()
    // {
    //     _producer = new ProducerBuilder<Null, string>(new ProducerConfig
    //     {
    //         BootstrapServers = "localhost:9092",
    //         Acks = Acks.All,
    //         MessageSendMaxRetries = 3
    //     }).Build();
    // }
    //
    // public async Task PublishAsync<T>(string topic, object message, CancellationToken cancellationToken = default) where T : class
    // {
    //     try
    //     {
    //         var result = await _producer.ProduceAsync(topic, new Message<Null, string>
    //         {
    //             Value = JsonConvert.SerializeObject(message)
    //         }, cancellationToken);
    //         
    //         Console.WriteLine($"Message '{result.Value}' delivered to {result.TopicPartitionOffset}");
    //     }
    //     catch (ProduceException<Null, string> e)
    //     {
    //         Console.WriteLine($"Error producing message: {e.Error.Reason}");
    //     }
    // }


    public async Task Publish2()
    {
        var services = new ServiceCollection();

        const string topicName = "test-topic";
        const string producerName = "say-hello";
 
        services.AddKafka(
            kafka => kafka
                .UseConsoleLog()
                .AddCluster(
                    cluster => cluster
                        .WithBrokers(new[] { "localhost:9092" })
                        .CreateTopicIfNotExists(topicName, 1, 1)
                        .AddProducer(
                            producerName,
                            producer => producer
                                .DefaultTopic(topicName)
                                .AddMiddlewares(m =>
                                    m.AddSerializer<JsonCoreSerializer>()
                                )
                        )
                )
        );

        var serviceProvider = services.BuildServiceProvider();

        var producer = serviceProvider
            .GetRequiredService<IProducerAccessor>()
            .GetProducer(producerName);
  
        for (var i = 0; i < 100; i++)
        {
            await producer.ProduceAsync(
                topicName,
                Guid.NewGuid().ToString(),
                new  { Text = i });

            Console.WriteLine("Message sent!");
        }
    }
}